#include <UTILS/SBuffer.h>
#include <PECAgent/PEContext.h>
#include <DPS/PH/DpsFunctions.h>
#include "UDOP_ReadDatem.h"
#include "iostream"
#include "fstream"
#include "utilities.h"
#include "analysis.h"

SPC_NAMESPACE_USE;
UTILS_NAMESPACE_USE;
using namespace DPS;
using namespace std;
using namespace estd;

/*
 * Function used to initialize the operator
 */
void UDOP_ReadDatem::initializeOperator() 
{
  // Get the paramter's value
  char str10[150];
  const DpsOPContext &context = this->getContext();
  const std::string &args = context.getCmdArguments();
  strcpy(str10, args.c_str());
  sscanf(str10,"%*s %*c%d%*c %*s %*c%d%*c",&num_ensemble,&ostep);
  // Create a user defined metric
  // opm.createMetric("intuples","Number of ingested tuples");
}

/*
 * Function used to finalize the operator
 */
void UDOP_ReadDatem::finalizeOperator() 
{
}

/*
 * Function used to process command line arguments
 */
void UDOP_ReadDatem::processCmdArgs(const string & /*args*/) 
{
  SPCDBG(L_INFO, "Process command line arguments...", "dpsop");
  // ArgContainer arg(args); // include "ArgContainer.h"
  // arg.argc and arg.argv are now available for use
  // do other initialization here
}


/*
 * Process function for input port 0
 */
void UDOP_ReadDatem::processInput0(const IPort0_t & tuple)
{
  SPCDBG(L_DEBUG, "Processing tuple from input port 0 " << tuple, "dpsop");
  int yr=0; int mo=0;
  int dy=0; int hr=0;
  short numobs=0;
  bool flag = true;	
  const short &tstep = (const short&) (tuple.get_ts() );

  dlist<float> obs_lat1 = DPS::makeFloatList();
  dlist<float> &obs_lat = obs_lat1;
  dlist<float> obs_lon1 = DPS::makeFloatList();
  dlist<float> &obs_lon = obs_lon1;
  dlist<double> y_obs1 = DPS::makeDoubleList();
  dlist<double> &y_obs = y_obs1;
  OPort0_t otuple;
  if (tstep == 1)
  {
	  // Initial guess
	  init_guess(num_ensemble);
	  // Read CONTROL file to obtain time		
	  get_time(&yr, &mo, &dy, &hr);
	  // Extract observations at this time point, and submit to down stream
	  obs_datem(yr, mo, dy, hr, obs_lat, obs_lon, y_obs, &numobs);
 
  if((y_obs.size() > 0) && (tstep%ostep == 0) )
  {
	flag = true;
  }
  else
  {
	  flag = false;
  }

  otuple.set_obs_lat(obs_lat);
  otuple.set_obs_lon(obs_lon);
  otuple.set_y_obs(y_obs);
  otuple.set_ts(tstep);
  otuple.set_flag(flag);
  otuple.set_numobs(numobs);
  }

  submit0(otuple);

  /*
   * OPort0_t otuple; submit0(otuple);
   * // updatePerformanceCounters(opm);
   */
}

/*
 * Process function for input port 1
 */
void UDOP_ReadDatem::processInput1(const IPort1_t & tuple)
{
  SPCDBG(L_DEBUG, "Processing tuple from input port 1 " << tuple, "dpsop");
  int yr=0; int mo=0;
  int dy=0; int hr=0;
  short numobs=0;
  bool flag = true;
  const short &tstep = (const short&) (tuple.get_ts() );

  dlist<float> obs_lat1 = DPS::makeFloatList();
  dlist<float> &obs_lat = obs_lat1;
  dlist<float> obs_lon1 = DPS::makeFloatList();
  dlist<float> &obs_lon = obs_lon1;
  dlist<double> y_obs1 = DPS::makeDoubleList();
  dlist<double> &y_obs = y_obs1;

  OPort0_t otuple;

  if (tstep > 1)
  {
  // Read CONTROL file to obtain time
  get_time(&yr, &mo, &dy, &hr);
  // Extract observations at this time point, and submit to down stream
  obs_datem(yr, mo, dy, hr, obs_lat, obs_lon, y_obs, &numobs);
  }

  if((y_obs.size() > 0) && (tstep%ostep == 0) )
  {
	  flag = true;
  }
  else
  {
	  flag = false;
  }

  //OPort0_t otuple;
  otuple.set_obs_lat(obs_lat);
  otuple.set_obs_lon(obs_lon);
  otuple.set_y_obs(y_obs);
  otuple.set_ts(tstep);
  otuple.set_flag(flag);
  otuple.set_numobs(numobs);

  submit0(otuple);
  /*
   * OPort0_t otuple; submit0(otuple);
   * // updatePerformanceCounters(opm);
   */
}

/*
 *Function used to read Datem file and extract observation at a time point
*/
void UDOP_ReadDatem::obs_datem(int yr, int mo, int dy, int hr, dlist<float> &obs_lat, dlist<float> &obs_lon, dlist<double> &y_obs, short *numobs)
{
	char str1[100];
	int iyr; int imo;
	int idy; int ishr; int idur;
	short k = 0;
	float ilat; float ilon;
	double icon;
	ifstream fin;

	fin.open("/home/shiming1/InfoSphereStreams/letkf-spade/hysplit/data/meas-t1.txt", ios::in);//TODO: Make this path not that absolute.
	fin.getline(str1, 100);
	fin.getline(str1, 100);

	while (!fin.eof())
	{
		fin>>iyr; iyr-=1900;// TODO: Make it valid for year after 2000
		fin>>imo;
		fin>>idy;
		fin>>ishr; ishr/=100;
		fin>>idur; idur/=100;
		fin>>ilat;
		fin>>ilon;
		fin>>icon;
		fin.getline(str1, 100);

		adjust_time(idur, &iyr, &imo, &idy, &ishr);
		if((iyr == yr) && (mo == imo) && (dy == idy) && (hr == ishr) )
		{
			obs_lat.push_back(ilat);
			obs_lon.push_back(ilon);
			icon *= 0.000000000001;
			y_obs.push_back(icon);
			k++;
		}
	}
	*numobs = k;
	fin.close();
}

/*
 * Punctuate function
 */
/*void UDOP_ReadDatem::processPunctuation(const Punctuation::Value & value, unsigned int input)
{
  SPCDBG(L_DEBUG, "Processing punctuation from input port " << input, "dpsop");
  // punctuate0(value);
}
*/

/*
 * Function used to notify the operator about the readiness of all ports (this
 * is useful when submitting tuples from within a context other than the process
 * functions of the operator).
 */ 
/*
void UDOP_ReadDatem::allPortsReady() 
{
} 
*/

/*
 * Terminate function, called asynchronously
 */
/*
void UDOP_ReadDatem::prepareToTerminate() 
{
}
*/

/*
 * Function used to update performance counters
 */
/*
static void updatePerformanceCounters(DpsOPMetrics & opm) 
{ 
  static long long ntuples = 0;
  // This is not the ideal way of using performance counters
  // Ideally one should store the result from getMetricByName, 
  // in a class member variable. For instance:
  //   DpsMetric & m = opm.getMetricByName("intuples");
  // m could be made a member variable of UDOP_ReadDatem
  // Then, the following is a more efficient way to use the counter:
  //   m.setValue(ntuples++);
  opm.getMetricByName("intuples").setValue(ntuples++);
}
*/

/*
 * Checkpointing function for saving state, called asynchronously 
 */
/*
void UDOP_ReadDatem::getCheckpoint(SBuffer & checkpoint) 
{
  SPCDBG(L_DEBUG, "Getting operator state...", "dpsop");
}
*/

/*
 * Checkpointing function for restoring state, called asynchronously
 */
/*
void UDOP_ReadDatem::restoreCheckpoint(SBuffer & checkpoint) 
{
  SPCDBG(L_DEBUG, "Restoring operator state...", "dpsop");
}
*/


